package transport.handler;


import hdfs.TransPort;
import io.netty.channel.*;
import io.netty.util.ReferenceCountUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import transport.ChannelUtils;
import transport.ChannelContext;

import java.io.FileNotFoundException;
import java.io.FileOutputStream;
import java.io.IOException;
import java.util.concurrent.ArrayBlockingQueue;

/**
 * Handle Request From Client
 */
public class OuterHandler extends WriteHandler {

    private static final Logger LOG = LoggerFactory.getLogger(OuterHandler.class);
    private Channel remoteChannel;
    private int blockInWrite = -1;
    private FileOutputStream writer;
    private ArrayBlockingQueue<ChannelContext> bridge;
    private ChannelContext remoteCtx;
    public OuterHandler(ArrayBlockingQueue<ChannelContext> bridge, String storagePath, EventLoopGroup aGroup, EventLoopGroup wGroup) throws FileNotFoundException {
        super(storagePath,aGroup,wGroup);
        try {
            remoteCtx = bridge.take();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        this.bridge = bridge;
//        System.out.println("OuterHandler"+System.identityHashCode(this.relayCtx));
    }

    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        super.channelActive(ctx);
        ctx.fireChannelActive();
        remoteCtx.setRelayChannel(ctx.channel());
        LOG.info("设置channel{}",remoteCtx.getRelayChannel());
        remoteChannel = remoteCtx.getRemoteChannel();
    }

    @Override
    public void channelInactive(ChannelHandlerContext ctx) throws IOException {
        LOG.info("channel {} closed", ctx.channel());
        writer.close();
        if(remoteChannel!=null){
            ChannelUtils.closeOnFlush(remoteChannel);
        }
    }

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws IOException {
        // act as a data transfer handler
        TransPort.WritePacketRequest req = (TransPort.WritePacketRequest) msg;
        LOG.info("{} 获取到Packet{}",ctx.channel().localAddress(),req.getSize());
        if(req.getSize()==0) return;
        int blockNum = req.getBlockNum();
        if(blockNum!=blockInWrite){
            if(writer!=null){
                writer.close();
            }
            writer = new FileOutputStream(storagePath+blockNum);
            blockInWrite = blockNum;
        }
        byte[] data = req.getData().toByteArray();
        if(remoteChannel!=null){
            if (remoteChannel.isActive()) {
                LOG.info("向{}写入{}",remoteChannel,req.getSize());
                remoteChannel.writeAndFlush(req).addListener(new ChannelFutureListener() {
                            @Override
                            public void operationComplete(ChannelFuture channelFuture) throws Exception {
                                if (!channelFuture.isSuccess()) {
                                    LOG.warn("write data to remote address occurs error {}", remoteChannel.remoteAddress(), channelFuture.cause());
                                    ChannelUtils.closeOnFlush(remoteChannel);
                                    ctx.close();
                                } else {
                                    ChannelUtils.tryToReadIfNeeded(ctx);
                                }
                            }
                });
            } else {
                ReferenceCountUtil.release(msg);
                ctx.close();
            }
        }else{
            TransPort.WritePacketRequest.Builder rep = TransPort.WritePacketRequest.newBuilder();
            rep.setSize(0);
            ctx.channel().writeAndFlush(rep.build()).addListener(new ChannelFutureListener() {
                @Override
                public void operationComplete(ChannelFuture channelFuture) throws Exception {
                    if (channelFuture.isSuccess()) {
                        ChannelUtils.tryToReadIfNeeded(ctx);
                    } else {
                        ChannelUtils.closeOnFlush(ctx.channel());
                    }
                }
            });
        }
        writer.write(data,0,req.getSize());
        writer.flush();

    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        LOG.warn("{}request handler occurs error{}",ctx.channel(), cause.getStackTrace());
        ctx.close();
        ChannelUtils.closeOnFlush(remoteChannel);
        writer.close();
    }

}
